package com.atguigu.tingshu.common.kafka.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.support.Acknowledgment;

/**
 * Kafka 消费者监听器示例
 * <p>
 * 注意：这是一个示例文件，实际使用时应该在各个业务模块中创建对应的监听器
 * <p>
 * 使用方式：
 * 1. 复制此类到你的业务模块中
 * 2. 修改 @KafkaListener 注解中的 topics 和 groupId
 * 3. 实现具体的业务逻辑
 * 4. 处理成功后调用 ack.acknowledge() 手动确认消息
 */
@Slf4j
// @Component  // 示例文件，不需要注册为 Bean，实际使用时取消注释
public class KafkaListenerExample {

    /**
     * 消费用户注册消息示例
     *
     * @param record 消息记录
     * @param ack    手动确认对象
     */
    // @KafkaListener(topics = KafkaConst.TOPIC_USER_REGISTER, groupId = KafkaConst.GROUP_USER)
    public void handleUserRegister(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            // 1. 获取消息内容
            String message = record.value();
            log.info("收到用户注册消息 - Topic: {}, Partition: {}, Offset: {}, Message: {}",
                    record.topic(), record.partition(), record.offset(), message);

            // 2. 解析消息（根据实际的消息类型进行解析）
            // UserInfo userInfo = JSON.parseObject(message, UserInfo.class);

            // 3. 幂等性校验（可选）
            // String messageId = record.headers().lastHeader("messageId").value().toString();
            // if (isMessageProcessed(messageId)) {
            //     log.warn("消息已处理过，跳过 - MessageId: {}", messageId);
            //     ack.acknowledge();
            //     return;
            // }

            // 4. 执行业务逻辑
            // processUserRegister(userInfo);

            // 5. 手动确认消息
            ack.acknowledge();
            log.info("用户注册消息处理成功 - Offset: {}", record.offset());

        } catch (Exception e) {
            log.error("处理用户注册消息失败 - Offset: {}, Error: {}",
                    record.offset(), e.getMessage(), e);
            // 注意：不调用 ack.acknowledge()，消息会重新消费
            // 如果需要跳过错误消息，可以调用 ack.acknowledge()
        }
    }

    /**
     * 消费订单支付成功消息示例
     *
     * @param record 消息记录
     * @param ack    手动确认对象
     */
    // @KafkaListener(topics = KafkaConst.TOPIC_ORDER_PAY_SUCCESS, groupId = KafkaConst.GROUP_ORDER)
    public void handleOrderPaySuccess(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            String message = record.value();
            log.info("收到订单支付成功消息 - Message: {}", message);

            // 解析消息并处理业务逻辑
            // OrderPaySuccessMessage payMessage = JSON.parseObject(message, OrderPaySuccessMessage.class);
            // processOrderPaySuccess(payMessage);

            // 手动确认
            ack.acknowledge();
            log.info("订单支付成功消息处理完成");

        } catch (Exception e) {
            log.error("处理订单支付成功消息失败", e);
        }
    }

    /**
     * 批量消费消息示例
     *
     * @param records 消息记录列表
     * @param ack     手动确认对象
     */
    // @KafkaListener(topics = KafkaConst.TOPIC_ALBUM_STAT_UPDATE, groupId = KafkaConst.GROUP_ALBUM)
    public void handleAlbumStatUpdateBatch(
            org.springframework.kafka.support.KafkaHeaders records,
            Acknowledgment ack) {
        try {
            log.info("批量消费专辑统计更新消息");

            // 批量处理消息
            // for (ConsumerRecord<String, String> record : records) {
            //     String message = record.value();
            //     // 处理单条消息
            // }

            // 批量确认
            ack.acknowledge();
            log.info("批量消息处理完成");

        } catch (Exception e) {
            log.error("批量处理消息失败", e);
        }
    }

    /**
     * 检查消息是否已处理（幂等性校验）
     *
     * @param messageId 消息ID
     * @return 是否已处理
     */
    private boolean isMessageProcessed(String messageId) {
        // 可以从 Redis 或数据库中查询消息是否已处理
        // return redisTemplate.hasKey("processed:message:" + messageId);
        return false;
    }

}

